大家好，我是君哥。

今天来分享 RocketMQ 的定时任务。通过这些定时任务，能让我们更加理解 RocketMQ 的消息处理机制和设计理念。

从 RocketMQ 4.9.4 的源代码上看，RocketMQ 的定时任务有很多，今天主要讲解一些核心的定时任务。

首先再来回顾一下 RocketMQ 的架构图：

Name Server 集群部署，但是节点之间并不会同步数据，因为每个节点都会保存完整的数据。因此单个节点挂掉，并不会对集群产生影响。

Broker 可以采用主从集群部署，实现多副本存储和高可用。每个 Broker 节点都要跟所有的 Name Server 节点建立长连接，定义注册 Topic 路由信息和发送心跳。

Producer 和 Consumer 跟 Name Server 的任意一个节点建立长连接，定期从 Name Server 拉取 Topic 路由信息。

Producer 和 Consumer 要跟 Name Server 建立连接，就必须首先获取 Name Server 地址。Producer 和 Consumer 采用定时任务每两分钟获取 Name Server 地址并更新本地缓存。代码如下：

Producer 和 Consumer 会定时从 Name Server 获取定时订阅信息，更新本地缓存，默认间隔是 30s（可以配置）。代码如下：

@Override public void run(){ try { MQClientInstance.this.updateTopicRouteInfoFromNameServer(); } catch (Exception e) { log.error("ScheduledTask updateTopicRouteInfoFromNameServer exception", e); } }}, 10, this.clientConfig.getPollNameServerInterval(), TimeUnit.MILLISECONDS);

Producer 和 Consumer 会从本地缓存的 Broker 列表中定时清除离线的 Broker，并且向 Broker 发送心跳，默认间隔是 30s（可以配置）。代码如下：

@Override public void run(){ try { MQClientInstance.this.cleanOfflineBroker(); MQClientInstance.this.sendHeartbeatToAllBrokerWithLock(); } catch (Exception e) { log.error("ScheduledTask sendHeartbeatToAllBroker exception", e); } }}, 1000, this.clientConfig.getHeartbeatBrokerInterval(), TimeUnit.MILLISECONDS);

消费者需要定时持久化 MessageQueue 的偏移量，默认每 5s 更新一次（可以配置）。

注意：集群模式需要向 Broker 发送持久化消息，因为集群模式偏移量保存在 Broker 端，而广播模式只需要把偏移量保存在消费者本地文件。代码如下：

@Override public void run(){ try { MQClientInstance.this.persistAllConsumerOffset(); } catch (Exception e) { log.error("ScheduledTask persistAllConsumerOffset exception", e); } }}, 1000 * 10, this.clientConfig.getPersistConsumerOffsetInterval(), TimeUnit.MILLISECONDS);

对于消费者采用推模式的情况，消费者会根据未消费的消息数量，定期更新核心线程数，默认每 1m 一次。

注意：在 4.9.4 这个版本，更新核心线程数的代码并没有实现，只是预留了接口。代码如下：

@Override public void run(){ try { MQClientInstance.this.adjustThreadPool(); } catch (Exception e) { log.error("ScheduledTask adjustThreadPool exception", e); } }}, 1, 1, TimeUnit.MINUTES);

Producer 和 Consumer 会定时扫描缓存在本地的请求，如果请求开始时间加超时时间（再加 1s）小于当前时间，则这个请求过期。通过定时任务（3s 一次）让过期请求失效，并且触发回调函数。

生产者发送消息后，会对成功失败的状态、花费时间进行记录，以此来计算吞吐量 TPS，响应时间 RT，代码如下：

executorService.scheduleAtFixedRate(new TimerTask() { private void printStats(){ if (snapshotList.size() >= 10) { doPrintStats(snapshotList, statsBenchmark, false); } }

@Override public void run(){ try { this.printStats(); } catch (Exception e) { e.printStackTrace(); } }}, 10000, 10000, TimeUnit.MILLISECONDS);

对于顺序消息，要保证同一个 MessageQueue 只能被同一个 Consumer 消费。消费者初始化的时候，会启动一个定时任务，定时（默认 20s，可以配置）地向 Broker 发送锁定消息，Broker 收到请求后，就会把 MessageQueue、group 和 clientId 进行绑定，这样其他客户端就不能从这个 MessageQueue 拉取消息。

代码如下：

注意：Broker 的加锁是有时效的（默认 60s，可以配置），过期后，有可能被其他 Consumer 进行消费。

Consumer 每秒会记录一次性能快照，比如消息从创建到消费花费的时间，消息从保存到消费花费的时间，接收到消息的总数量，失败总数量。代码如下：

上面记录了性能快照后，Consumer 会每隔 10s 进行性能参数计算和打印。代码如下：

final long consumeTps = (long) (((end[1] - begin[1]) / (double) (end[0] - begin[0])) * 1000L); final double averageB2CRT = (end[2] - begin[2]) / (double) (end[1] - begin[1]); final double averageS2CRT = (end[3] - begin[3]) / (double) (end[1] - begin[1]); final long failCount = end[4] - begin[4]; final long b2cMax = statsBenchmarkConsumer.getBorn2ConsumerMaxRT().get(); final long s2cMax = statsBenchmarkConsumer.getStore2ConsumerMaxRT().get();

statsBenchmarkConsumer.getBorn2ConsumerMaxRT().set(0); statsBenchmarkConsumer.getStore2ConsumerMaxRT().set(0);

System.out.printf("Current Time: %s TPS: %d FAIL: %d AVG(B2C) RT(ms): %7.3f AVG(S2C) RT(ms): %7.3f MAX(B2C) RT(ms): %d MAX(S2C) RT(ms): %d%n", System.currentTimeMillis(), consumeTps, failCount, averageB2CRT, averageS2CRT, b2cMax, s2cMax ); }}

通过性能参数的日志输出，可以很方便的对 RocketMQ 的消费者进行监控。

消费者会定期检查本地拉取的消息列表，如果列表中的消息已经过期（默认 15 分钟过期，可以配置），则把过期消息再次发送给 Broker，然后从本地消息列表删除。代码如下：

@Override public void run(){ try { cleanExpireMsg(); } catch (Throwable e) { log.error("scheduleAtFixedRate cleanExpireMsg exception", e); } }

}, this.defaultMQPushConsumer.getConsumeTimeout(), this.defaultMQPushConsumer.getConsumeTimeout(), TimeUnit.MINUTES);

消费者会每隔 30s 向 NameServer 拉取 MessageQueue 信息，然后跟本地保存的进行比较，如果不一致，则更新本地缓存信息。代码如下：

Broker 端会对状态进行采用，比如一个 Topic、MessageQueue、Group 总共发送了多少条消息，Topic 总共发送的消息大小。Broker 会对这些状态按照秒、分钟、小时为单位进行采样并且定时打印，这里一共有 6 个定时任务。比如下面是按照秒进行采样的定时任务：

Broker 读取消息时会记录消息从保存磁盘到被读取的时间差并定时打印。定时任务代码如下：

Broker 会定时持久化消费偏移量、Topic 配置、定阅组配置等，默认 10s 一次（可以配置）。代码如下：

@Override public void run(){ try { if (started.get()) { ScheduleMessageService.this.persist(); } } catch (Throwable e) { log.error("scheduleAtFixedRate flush exception", e); } }}, 10000, this.defaultMessageStore.getMessageStoreConfig().getFlushDelayOffsetInterval(), TimeUnit.MILLISECONDS);

Broker 会定时扫描缓存在本地的请求，如果请求开始时间加超时时间（再加 1s）小于当前时间，则这个请求过期。通过定时任务（3s 一次）让过期请求失效，并且触发回调函数。

@Override public void run(){ try { NettyRemotingServer.this.scanResponseTable(); } catch (Throwable e) { log.error("scanResponseTable exception", e); } }}, 1000 * 3, 1000);

消费者可能会向 Broker 注册 filterClass 用来过滤消息。Broker 收到消费者注册的 filterClass 后会用定时任务来创建 FilterServer。代码如下：

这样消费者拉取消息时首先从 FilterServer 拉取消息，FilterServer 从 Broker 拉取消息后进行过滤，只把消费者感兴趣的消息返回给消费者。一个 Broker 可以有多个 FilterServer。如下图：

Broker 每天会记录前一天收发消息的总数量，定时任务如下（period 是 1 天）：

Broker 默认每隔 5s（可以配置） 会持久化一次消息的 Offset，代码如下：

上面提到过，消费者可能会向 Broker 注册 filterClass，Broker 解析消费者注册的 filterClass 后，会把解析后的 FilterData 持久化到文件，代码如下：

当消费者读取消息缓慢时，Broker 为了保护自己，会把这个消费者设置为不允许读取的状态，这样这个消费组就不能再拉取消息了，代码如下：

Broker 会每隔 1s 打印一次水位，包括发送消息的延迟、接收消息的延迟、事务消息的延迟、查询消息的延迟，代码如下：

Broker 会定时打印最新的消息 Offset 和已经分发给 MessageQueue 和 Index 索引的 Offset 差距，代码如下：

Broker 会定期获取 NameServer 的地址，并更新本地缓存，代码如下：

@Override public void run(){ try { BrokerController.this.brokerOuterAPI.fetchNameServerAddr(); } catch (Throwable e) { log.error("ScheduledTask fetchNameServerAddr exception", e); } }}, 1000 * 10, 1000 * 60 * 2, TimeUnit.MILLISECONDS);

Broker 会定时打印 master 节点和 slave 节点消息 Offset 的差距，代码如下：

Broker 会定时向（默认 30s，可配置，最高不超过 60s）所有 NameServer 发送注册消息，代码如下：

@Override public void run(){ try { BrokerController.this.registerBrokerAll(true, false, brokerConfig.isForceRegister()); } catch (Throwable e) { log.error("registerBrokerAll Exception", e); } }}, 1000 * 10, Math.max(10000, Math.min(brokerConfig.getRegisterNameServerPeriod(), 60000)), TimeUnit.MILLISECONDS);

Broker 的 Master 节点会每间隔 10s 向 Slave 节点同步数据，包括 Topic 配置、消费偏移量、延迟偏移量、消费组配置，代码如下：

Broker 会周期性（默认 10s，可以配置）地执行删除任务，删除过期的 CommitLog 文件和 ConsumeQueue 文件，代码如下：

Broker 会每隔 10 分钟检查 CommitLog 文件和 ConsumeQueue 文件，用当前文件的最小（起始） Offset 减去上一个文件最小（起始） Offset，如果不等于一个文件的大小，就说明文件存在问题。代码如下：

Broker 会每隔 1s 记录所有存活线程的堆栈映射信息，前提是 debugLockEnable 开关配置是打开的。代码如下：